package com.tdy.cdc.app.dwd;

import com.tdy.cdc.app.BaseSqlApp;
import com.tdy.cdc.common.Constant;
import com.tdy.cdc.util.FlinkSinkUtil;
import com.tdy.cdc.util.FlinkSourceUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @author NanHuang
 * @Date 2023/3/14
 */
public class App16_DwdMhsIptPreoperationSummary extends BaseSqlApp {
    public static void main(String[] args) {
        new App16_DwdMhsIptPreoperationSummary().init(
                5016,
                "App16_DwdMhsIptPreoperationSummary",
                2
        );
    }

    @Override
    protected void invoke(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        // 1 创建源表
        createSourceTables(tableEnv);
        // 2 join
        joinTables(tableEnv);
        // 3 写入Kafka
        writeToKafka(tableEnv);
    }

    private void writeToKafka( StreamTableEnvironment tableEnv) {
        // create kafka mappings
        tableEnv.executeSql("CREATE TABLE target ("+
                "    RECORD_ID                               STRING    "+
                "    ,HOSPITAL_NAME                          STRING    "+
                "    ,HOSPITAL_CODE                          STRING    "+
                "    ,AGE_UNIT                               STRING    "+
                "    ,AGE                                    STRING    "+
                "    ,SEX_NAME                               STRING    "+
                "    ,SEX_CODE                               STRING    "+
                "    ,ID_CARD_NUMBER                         STRING    "+
                "    ,PATIENT_VISIT_CATEGORY_CODE            STRING    "+
                "    ,PATIENT_VISIT_CATEGORY                 STRING    "+
                "    ,INPATIENT_VISIT_FLOW_ID                STRING    "+
                "    ,INPATIENT_NO                           STRING    "+
                "    ,PATIENT_ID                             STRING    "+
                "    ,PATIENT_NAME                           STRING    "+
                "    ,DOMAIN_ID                              STRING    "+
                "    ,VISIT_DATE                             STRING    "+
                "    ,VISIT_TIMES                            STRING    "+
                "    ,SUMMARY_DATETIME                       STRING    "+
                "    ,AUTHOR_CODE                            STRING    "+
                "    ,AUTHOR_NAME                            STRING    "+
                "    ,CUSTODIAN_CODE                         STRING    "+
                "    ,CUSTODIAN_NAME                         STRING    "+
                "    ,OPER_DOCTOR_SIGN_DATETIME              STRING    "+
                "    ,OPER_DOCTOR_ID                         STRING    "+
                "    ,OPER_DOCTOR_SIGN_NAME                  STRING    "+
                "    ,ELEC_SIGN_DATETIME                     STRING    "+
                "    ,DOCTOR_ID                              STRING    "+
                "    ,DOCTOR_SIGN_NAME                       STRING    "+
                "    ,LINKMAN_PHONE                          STRING    "+
                "    ,LINKMAN_NAME                           STRING    "+
                "    ,ADMIT_DATETIME                         STRING    "+
                "    ,DISCHARGE_DATETIME                     STRING    "+
                "    ,BED_NO                                 STRING    "+
                "    ,BED_NAME                               STRING    "+
                "    ,ROOM_NO                                STRING    "+
                "    ,ROOM_NAME                              STRING    "+
                "    ,WARD_ID                                STRING    "+
                "    ,WARD_NAME                              STRING    "+
                "    ,DEPT_ID                                STRING    "+
                "    ,DEPT_NAME                              STRING    "+
                "    ,MEDICAL_SUMMARY                        STRING    "+
                "    ,DISEASE_DIAGNOSIS_CODE                 STRING    "+
                "    ,PRE_OPER_DIAG_ICD_NAME                 STRING    "+
                "    ,DIAG_BASIS                             STRING    "+
                "    ,ALLERGIC_HISTORY_FLAG                  STRING    "+
                "    ,ALLERGIC_HISTORY                       STRING    "+
                "    ,ASSIST_EXAM_RESULT                     STRING    "+
                "    ,OPER_INDICATION                        STRING    "+
                "    ,OPER_CONTRAINDICATION                  STRING    "+
                "    ,SURGICAL_INDICATION                    STRING    "+
                "    ,CONSULTATION_OPINION                   STRING    "+
                "    ,INTEND_OPER_CM3_CODE                   STRING    "+
                "    ,INTEND_OPER_CM3_NAME                   STRING    "+
                "    ,INTEND_OPER_SITE_NAME                  STRING    "+
                "    ,INTEND_OPER_DATETIME                   STRING    "+
                "    ,INTEND_ANES_METHOD_CODE                STRING    "+
                "    ,INTEND_ANES_METHOD_NAME                STRING    "+
                "    ,CONSIDERATIONS                         STRING    "+
                "    ,OPER_KEY_POINT                         STRING    "+
                "    ,PRE_OPER_PREPARATION                   STRING    " +
                "    ,VISIT_ID                               STRING    "+
                "    ,PRIMARY KEY (RECORD_ID) NOT ENFORCED             "+
                ") "+ FlinkSinkUtil.getUpsertKafkaWith(Constant.TOPIC_DWD_MHS_IPT_PREOPERATION_SUMMARY));


        // load data
        tableEnv.executeSql("insert into target select * from join_result");
    }

    private void joinTables(StreamTableEnvironment tableEnv) {
        // 1 设置数据的TTL（暂时10min）
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(30 * 60));
        // 2 join操作
        Table joinResult = tableEnv.sqlQuery("select a.unique_id    AS    RECORD_ID                     "+
                "    ,a.hospital_name            AS       HOSPITAL_NAME                                              "+
                "    ,a.hospital_code            AS       HOSPITAL_CODE                                              "+
                "    ,a.age_unit                 AS       AGE_UNIT                                                   "+
                "    ,a.age                      AS       AGE                                                        "+
                "    ,a.sex                      AS       SEX_NAME                                                   "+
                "    ,a.sex_code                 AS       SEX_CODE                                                   "+
                "    ,a.identify_no              AS       ID_CARD_NUMBER                                             "+
                "    ,a.patient_typecode         AS       PATIENT_VISIT_CATEGORY_CODE                                "+
                "    ,a.patient_type             AS       PATIENT_VISIT_CATEGORY                                     "+
                "    ,a.visit_id                 AS       INPATIENT_VISIT_FLOW_ID                                    "+
                "    ,a.inpatient                AS       INPATIENT_NO                                               "+
                "    ,a.patient_id               AS       PATIENT_ID                                                 "+
                "    ,a.patient_name             AS       PATIENT_NAME                                               "+
                "    ,a.domain_id                AS       DOMAIN_ID                                                  "+
                "    ,a.visit_date               AS       VISIT_DATE                                                 "+
                "    ,a.visit_times              AS       VISIT_TIMES                                                "+
                "    ,a.effective_time           AS       SUMMARY_DATETIME                                           "+
                "    ,a.author_code              AS       AUTHOR_CODE                                                "+
                "    ,a.author_name              AS       AUTHOR_NAME                                                "+
                "    ,a.custodian_code           AS       CUSTODIAN_CODE                                             "+
                "    ,a.custodian_name           AS       CUSTODIAN_NAME                                             "+
                "    ,a.surgeon_signdate         AS       OPER_DOCTOR_SIGN_DATETIME                                  "+
                "    ,a.surgeon_code             AS       OPER_DOCTOR_ID                                             "+
                "    ,a.surgeon_name             AS       OPER_DOCTOR_SIGN_NAME                                      "+
                "    ,a.doctor_signdate          AS       ELEC_SIGN_DATETIME                                         "+
                "    ,a.doctor_code              AS       DOCTOR_ID                                                  "+
                "    ,a.doctor_name              AS       DOCTOR_SIGN_NAME                                           "+
                "    ,a.contact_number           AS       LINKMAN_PHONE                                              "+
                "    ,a.contact_name             AS       LINKMAN_NAME                                               "+
                "    ,a.admission_date           AS       ADMIT_DATETIME                                             "+
                "    ,a.out_date                 AS       DISCHARGE_DATETIME                                         "+
                "    ,a.bed_no                   AS       BED_NO                                                     "+
                "    ,a.bed_name                 AS       BED_NAME                                                   "+
                "    ,a.ward_id                  AS       ROOM_NO                                                    "+
                "    ,a.ward_name                AS       ROOM_NAME                                                  "+
                "    ,a.wards_id                 AS       WARD_ID                                                    "+
                "    ,a.wards_name               AS       WARD_NAME                                                  "+
                "    ,a.dept_code                AS       DEPT_ID                                                    "+
                "    ,a.dept_name                AS       DEPT_NAME                                                  "+
                "    ,a.record_summary           AS       MEDICAL_SUMMARY                                            "+
                "    ,case when b.diag_type='术前诊断' then b.diag_code else '' end  AS  DISEASE_DIAGNOSIS_CODE      "+
                "    ,case when b.diag_type='术前诊断' then b.diag_name else '' end  AS  PRE_OPER_DIAG_ICD_NAME      "+
                "    ,a.diagnostic_basis         AS       DIAG_BASIS                                                 "+
                "    ,a.allergy_history_mark     AS       ALLERGIC_HISTORY_FLAG                                      "+
                "    ,a.allergy_history          AS       ALLERGIC_HISTORY                                           "+
                "    ,a.item_result              AS       ASSIST_EXAM_RESULT                                         "+
                "    ,a.oper_indication          AS       OPER_INDICATION                                            "+
                "    ,a.contraindication         AS       OPER_CONTRAINDICATION                                      "+
                "    ,a.oper_surgical            AS       SURGICAL_INDICATION                                        "+
                "    ,a.consultation_opinion     AS       CONSULTATION_OPINION                                       "+
                "    ,a.oper_code                AS       INTEND_OPER_CM3_CODE                                       "+
                "    ,a.oper_name                AS       INTEND_OPER_CM3_NAME                                       "+
                "    ,a.target_name              AS       INTEND_OPER_SITE_NAME                                      "+
                "    ,a.oper_date                AS       INTEND_OPER_DATETIME                                       "+
                "    ,a.anes_method_code         AS       INTEND_ANES_METHOD_CODE                                    "+
                "    ,a.anes_name                AS       INTEND_ANES_METHOD_NAME                                    "+
                "    ,a.needing_attention        AS       CONSIDERATIONS                                             "+
                "    ,a.oper_key_point           AS       OPER_KEY_POINT                                             "+
                "    ,a.pre_preparation          AS       PRE_OPER_PREPARATION                                       " +
                "    ,visit_id AS VISIT_ID "+
                " from hdsd00_14_04 a                                                                                "+
                " left join tdy_list_diag b                                                                          "+
                " on a.unique_id = b.unique_id and a.xds_type = b.xds_type                                         ");
        tableEnv.createTemporaryView("join_result",joinResult);

    }

    private void createSourceTables(StreamTableEnvironment tableEnv) {
        // create table : hdsd00_14_04
        tableEnv.executeSql("create table hdsd00_14_04 ( "+
                "pk                     string," +
                "upload_time            string," +
                "status                 string," +
                "empi                   string," +
                "encounter_id           string," +
                "visit_date             string," +
                "patient_id             string," +
                "patient_domain         string," +
                "visit_domain           string," +
                "visit_id               string," +
                "visit_times            string," +
                "unique_id              string," +
                "xds_type               string," +
                "xds_name               string," +
                "effective_time         string," +
                "xds_version            string," +
                "domain_id              string," +
                "patient_type           string," +
                "patient_typecode       string," +
                "inpatient              string," +
                "identify_no            string," +
                "patient_name           string," +
                "sex_code               string," +
                "sex                    string," +
                "age                    string," +
                "age_unit               string," +
                "write_time             string," +
                "author_code            string," +
                "author_name            string," +
                "custodian_code         string," +
                "custodian_name         string," +
                "surgeon_signdate       string," +
                "surgeon_code           string," +
                "surgeon_name           string," +
                "doctor_signdate        string," +
                "doctor_code            string," +
                "doctor_name            string," +
                "contact_number         string," +
                "contact_name           string," +
                "admission_date         string," +
                "out_date               string," +
                "bed_no                 string," +
                "bed_name               string," +
                "ward_id                string," +
                "ward_name              string," +
                "wards_id               string," +
                "wards_name             string," +
                "dept_code              string," +
                "dept_name              string," +
                "hospital_code          string," +
                "hospital_name          string," +
                "record_summary         string," +
                "diagnostic_basis       string," +
                "allergy_history_mark   string," +
                "allergy_history        string," +
                "item_result            string," +
                "oper_indication        string," +
                "contraindication       string," +
                "oper_surgical          string," +
                "consultation_opinion   string," +
                "oper_code              string," +
                "oper_name              string," +
                "target_name            string," +
                "oper_date              string," +
                "anes_method_code       string," +
                "anes_name              string," +
                "needing_attention      string," +
                "oper_key_point         string," +
                "pre_preparation        string" +
                ")" + FlinkSourceUtil.getKafkaWith("hdsd00_14_04","App16_DwdMhsIptPreoperationSummary"));

        // create table : tdy_list_diag
        super.readTdyListDiag(tableEnv,"App16_DwdMhsIptPreoperationSummary");
    }
}
